package com.bole.reactivestream;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @author: 叶子
 * @date: 2022/04/10 20:24
 **/
public class ReactiveDemo2 {
    public static void main(String[] args) {

        //1.建立发布者
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        //2.创建处理器
        ReactiveProcessor processor = new ReactiveProcessor();
        //3.发布者与处理器建立订阅关系
        publisher.subscribe(processor);
        //4.创建订阅者
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println("建立订阅关系");
                this.subscription = subscription;
                this.subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                System.out.println("最终的subscriber接收到数据："+item);
                this.subscription.request(10);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("出现异常");
                throwable.printStackTrace();
                this.subscription.cancel();
            }

            @Override
            public void onComplete() {
                System.out.println("数据接收完成");
            }
        };
        //5.中间处理器与订阅者建立订阅关系
        processor.subscribe(subscriber);
        for (int i = 0; i < 500; i++) {
            System.out.println("发布数据=" + i);
            publisher.submit("bole" + i);
        }
        publisher.close();
        try {
            Thread.currentThread().join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}